package com.ww.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object RDDTry01 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("try01")
    val sc = new SparkContext(conf)
    val dataRDD: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 5, 5, 5, 9))


    dataRDD.filter(_ > 3).collect().foreach(print)
    println("\n==========================去重复==================================")

    dataRDD.map((_, 1)).reduceByKey(_ + _).map(_._1).foreach(print)
    println()
    // dataRDD.map((_,1)).reduceByKey(_+_).map((_._1,1)).foreach(print)
    dataRDD.distinct().foreach(print)
    println("\n==========================数据集合并==================================")

    val d1 = sc.parallelize(List(1, 2, 3, 4, 5, 5, 5))

    val d2 = sc.parallelize(List(3, 4, 5, 5, 5, 5, 9))

    val d3: RDD[Int] = d1.union(d2)
    println(d3.partitions.length)
    d3.foreach(print)
    println("\n==========================笛卡尔积==================================")
    val d4 = d1.cartesian(d2)
    d4.foreach(print)

    println("\n==========================交集==================================")
    d1.intersection(d2).foreach(print)
    println("\n==========================差集==================================")
    d1.subtract(d2).foreach(print)
    println("\n==========================cogroup==================================")
    val kv1: RDD[(String, Int)] = sc.parallelize(List(
      ("zhangsan", 11),
      ("zhangsan", 12),
      ("lisi", 13),
      ("wangwu", 14)
    ))
    val kv2: RDD[(String, Int)] = sc.parallelize(List(
      ("zhangsan", 21),
      ("zhangsan", 22),
      ("lisi", 23),
      ("zhaoliu", 28)
    ))
    kv1.cogroup(kv2).foreach(println)
    println("\n==========================join==================================")
    kv1.join(kv2).foreach(println)
    println("\n==========================left out join==================================")
    kv1.leftOuterJoin(kv2).foreach(println)
    println("\n==========================right out join==================================")
    kv1.rightOuterJoin(kv2).foreach(println)
    println("\n==========================full out join==================================")
    kv1.fullOuterJoin(kv2).foreach(println)
  }
}
